package app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import util.GmallSinkUtil;
import util.MyCommonUtils;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;

public class DwdLogAppFollow extends BaseAppFollow {
    public static void main(String[] args) {
        new DwdLogAppFollow().init(3001, 1, "DwdLogAppFollow", Constant.TOPIC_ODS_LOG, "DwdLogAppFollow");
    }

    @Override
    public void run(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        //1.识别新老用户
        /*① 事件时间
         * ② 用户分组
         * ③ 用户开窗,去第一个窗找登陆处女秀
         * ④ 如何判别第一个窗,使用键控状态*/

        SingleOutputStreamOperator<JSONObject> distinguishNewAndOldStream = distinguishNewAndOld(dataStreamSource);
        //2. log日志进行分流
        HashMap<String, DataStream<JSONObject>> threeStreamMap = splitStream(distinguishNewAndOldStream);
        DataStream<JSONObject> startStream = threeStreamMap.get("start");
        DataStream<JSONObject> pageStream = threeStreamMap.get("page");
        DataStream<JSONObject> displayStream = threeStreamMap.get("display");
        //3. 三类日志写入kafka不同topic,构建dwd层
        GmallSinkUtil.sendToKafka(pageStream.map(JSONAware::toJSONString), Constant.TOPIC_DWD_PAGE_LOG);
        GmallSinkUtil.sendToKafka(startStream.map(JSONAware::toJSONString), Constant.TOPIC_DWD_START_LOG);
        GmallSinkUtil.sendToKafka(displayStream.map(JSONAware::toJSONString), Constant.TOPIC_DWD_DISPLAY_LOG);
    }

    private HashMap<String, DataStream<JSONObject>> splitStream(SingleOutputStreamOperator<JSONObject> distinguishNewAndOldStream) {
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("displayTag") {
        };
        OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("pageTag") {
        };

        SingleOutputStreamOperator<JSONObject> startStream = distinguishNewAndOldStream.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject value,
                                       Context ctx,
                                       Collector<JSONObject> out) throws Exception {
                if (value.getJSONObject("start") != null) {
                    out.collect(value);
                } else {
                    JSONObject page = value.getJSONObject("page");
                    if (page != null) {
                        ctx.output(pageTag, value);
                    }
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            display.put("ts", value.getLong("ts"));
                            display.put("page_id", page.getString("page_id"));
                            display.putAll(value.getJSONObject("common"));
                            ctx.output(displayTag, display);
                        }
                    }
                }
            }
        });
        DataStream<JSONObject> displayStream = startStream.getSideOutput(displayTag);
        DataStream<JSONObject> pageStream = startStream.getSideOutput(pageTag);

        HashMap<String, DataStream<JSONObject>> threeStreamMap = new HashMap<>();
        threeStreamMap.put("start", startStream);
        threeStreamMap.put("page", pageStream);
        threeStreamMap.put("display", displayStream);
        return threeStreamMap;
    }


    private SingleOutputStreamOperator<JSONObject> distinguishNewAndOld(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource
            .map(JSON::parseObject)
            .assignTimestampsAndWatermarks(WatermarkStrategy
                                               .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                               .withTimestampAssigner((obj, ts) -> obj.getLong("ts")))
            .keyBy(obj -> obj.getJSONObject("common").getString("mid"))
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {

                private ValueState<Long> firstState;

                @Override
                public void process(String s,
                                    Context ctx, Iterable<JSONObject> eles,
                                    Collector<JSONObject> out) throws Exception {
                    if (firstState.value() == null) {
                        ArrayList<JSONObject> list = MyCommonUtils.iterToList(eles);
                        //如果状态为空,说明这是第一个窗,对迭代器里面数据进行排序取第一个,迭代器无法排序,创建工具类转换为数组
                        list.sort(Comparator.comparing(o -> o.getLong("ts")));

                        //遍历每一个数据标记isnew
                        for (int i = 0; i < list.size(); i++) {
                            JSONObject common = list.get(i).getJSONObject("common");
                            if (i == 0) {
                                common.put("is_new", 1);
                                firstState.update(list.get(i).getLong("ts"));
                            } else {
                                common.put("is_new", 0);
                            }
                            out.collect(list.get(i));
                        }
                    } else {
                        for (JSONObject ele : eles) {
                            ele.getJSONObject("common").put("is_new", 0);
                            out.collect(ele);
                        }
                    }
                }

                @Override
                public void open(Configuration parameters) throws Exception {
                    firstState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("firstState", Long.class));
                }
            });
    }


}
